use std::sync::{Arc, Mutex};
use tauri::{AppHandle, Emitter, State};
use tokio::sync::mpsc;
use tokio_tungstenite::{connect_async, tungstenite::Message as WsMessage};
use http::{Request, HeaderValue};
use futures_util::{SinkExt, StreamExt};
use serde::{Deserialize, Serialize};
use base64::{Engine as _, engine::general_purpose};

/// WebSocket 连接状态
#[derive(Debug, Clone)]
pub struct WebSocketState {
    pub sender: Option<mpsc::UnboundedSender<WsMessage>>,
    pub subscription_id: Option<String>,
}

impl Default for WebSocketState {
    fn default() -> Self {
        Self {
            sender: None,
            subscription_id: None,
        }
    }
}

/// 消息事件
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MessageEvent {
    pub message: serde_json::Value,
}

/// 连接状态事件
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConnectionEvent {
    pub status: String, // "connected" | "disconnected" | "error"
    pub message: Option<String>,
}

/// 连接 WebSocket
#[tauri::command]
pub async fn connect_websocket(
    subscription_id: String,
    ws_url: Option<String>,
    token: Option<String>,
    app: AppHandle,
    state: State<'_, Arc<Mutex<WebSocketState>>>,
) -> Result<(), String> {
    // 如果已经连接，先断开
    {
        let needs_disconnect = {
            let state_guard = state.lock().unwrap();
            state_guard.sender.is_some()
        };
        if needs_disconnect {
            disconnect_websocket_internal(&app, &state).await?;
        }
    }

    let default_url = format!("ws://127.0.0.1:3001/ws/{}", subscription_id);
    let url_str = ws_url.unwrap_or(default_url);

    // 发送连接中事件
    app.emit(
        "websocket-status",
        ConnectionEvent {
            status: "connecting".to_string(),
            message: Some("正在连接...".to_string()),
        },
    )
    .map_err(|e| format!("发送事件失败: {}", e))?;

    // 构建 WebSocket 请求，手动添加所有必需的 WebSocket 握手头部
    // 因为 connect_async 在手动构建 Request 时可能不会自动添加这些头部
    let url = url_str.parse::<url::Url>()
        .map_err(|e| format!("无效的 WebSocket URL: {}", e))?;
    
    // 从 URL 中提取 Host 头部值
    // 对于标准端口（HTTP 80, HTTPS 443），Host 头部不应该包含端口号
    // 对于非标准端口，需要包含端口号
    let host = match (url.host_str(), url.port(), url.scheme()) {
        (Some(host), Some(port), _) => {
            // 检查是否是标准端口
            let is_standard_port = match url.scheme() {
                "ws" | "http" => port == 80,
                "wss" | "https" => port == 443,
                _ => false,
            };
            if is_standard_port {
                host.to_string()
            } else {
                format!("{}:{}", host, port)
            }
        },
        (Some(host), None, _) => host.to_string(),
        (None, _, _) => return Err("URL 中缺少 host".to_string()),
    };
    
    // 生成 Sec-WebSocket-Key（16 字节随机数，Base64 编码）
    use rand::RngCore;
    let mut key_bytes = [0u8; 16];
    rand::thread_rng().fill_bytes(&mut key_bytes);
    let sec_websocket_key = general_purpose::STANDARD.encode(&key_bytes);
    
    // 构建请求，手动添加所有必需的 WebSocket 握手头部
    let mut request_builder = Request::builder()
        .uri(url.as_str())
        .header("Host", &host)
        .header("Upgrade", "websocket")
        .header("Connection", "Upgrade")
        .header("Sec-WebSocket-Key", &sec_websocket_key)
        .header("Sec-WebSocket-Version", "13");
    
    // 添加自定义头部（Authorization）
    if let Some(t) = &token {
        let auth_header = format!("Bearer {}", t);
        request_builder = request_builder.header(
            "Authorization",
            HeaderValue::from_str(&auth_header)
                .map_err(|e| format!("无效的 Authorization header: {}", e))?,
        );
    }
    
    let request = request_builder
        .body(())
        .map_err(|e| format!("构建 WebSocket 请求失败: {}", e))?;

    // 连接 WebSocket
    let (ws_stream, _) = connect_async(request)
        .await
        .map_err(|e| format!("WebSocket连接失败: {}", e))?;

    let (mut write, mut read) = ws_stream.split();

    // 创建消息通道
    let (tx, mut rx) = mpsc::unbounded_channel::<WsMessage>();

    // 更新状态
    {
        let mut state_guard = state.lock().unwrap();
        state_guard.sender = Some(tx.clone());
        state_guard.subscription_id = Some(subscription_id.clone());
    }

    // 发送连接成功事件
    app.emit(
        "websocket-status",
        ConnectionEvent {
            status: "connected".to_string(),
            message: Some("连接成功".to_string()),
        },
    )
    .map_err(|e| format!("发送事件失败: {}", e))?;

    // 发送消息任务（包括 ping）
    let write_task_app = app.clone();
    let write_task_state = Arc::clone(&state.inner());
    let mut ping_interval = tokio::time::interval(tokio::time::Duration::from_secs(30));
    ping_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    
    tokio::spawn(async move {
        loop {
            tokio::select! {
                _ = ping_interval.tick() => {
                    // 定期发送 ping 保持连接活跃
                    if write.send(WsMessage::Ping(vec![])).await.is_err() {
                        // 发送 ping 失败，连接可能已断开
                        let mut state_guard = write_task_state.lock().unwrap();
                        state_guard.sender = None;
                        state_guard.subscription_id = None;
                        
                        let _ = write_task_app.emit(
                            "websocket-status",
                            ConnectionEvent {
                                status: "error".to_string(),
                                message: Some("发送 ping 失败，连接可能已断开".to_string()),
                            },
                        );
                        break;
                    }
                }
                msg_opt = rx.recv() => {
                    match msg_opt {
                        Some(msg) => {
                            if write.send(msg).await.is_err() {
                                // 发送失败，更新状态并通知前端
                                let mut state_guard = write_task_state.lock().unwrap();
                                state_guard.sender = None;
                                state_guard.subscription_id = None;
                                
                                let _ = write_task_app.emit(
                                    "websocket-status",
                                    ConnectionEvent {
                                        status: "error".to_string(),
                                        message: Some("发送消息失败，连接可能已断开".to_string()),
                                    },
                                );
                                break;
                            }
                        }
                        None => {
                            // 通道已关闭
                            break;
                        }
                    }
                }
            }
        }
    });

    // 接收消息任务
    let app_clone = app.clone();
    let state_arc = Arc::clone(&state.inner());
    let subscription_id_clone = subscription_id.clone();
    tokio::spawn(async move {
        let mut _last_message_time = std::time::Instant::now();
        let mut consecutive_errors = 0;
        const MAX_CONSECUTIVE_ERRORS: u32 = 5;
        
        while let Some(msg_result) = read.next().await {
            match msg_result {
                Ok(WsMessage::Text(text)) => {
                    _last_message_time = std::time::Instant::now();
                    consecutive_errors = 0;
                    eprintln!("收到 WebSocket 文本消息: {}", text);
                    // 解析 JSON 消息
                    match serde_json::from_str::<serde_json::Value>(&text) {
                        Ok(json) => {
                            let chat_type = json.get("chat_type");
                            let is_group_message = chat_type.and_then(|v| v.as_i64()).map(|v| v == 2).unwrap_or(false);
                            
                            eprintln!("成功解析 JSON 消息，chat_type: {:?}, 是否为群组消息: {}", chat_type, is_group_message);
                            eprintln!("消息字段检查: from_user_id={:?}, to_user_id={:?}, message_id={:?}", 
                                json.get("from_user_id"), 
                                json.get("to_user_id"),
                                json.get("message_id"));
                            
                            if is_group_message {
                                eprintln!("🔵 检测到群组消息: to_user_id={:?}, chat_type={:?}", 
                                    json.get("to_user_id"), 
                                    chat_type);
                            }
                            
                            // 发送消息事件到前端
                            if let Err(e) = app_clone.emit(
                                "websocket-message",
                                MessageEvent { message: json.clone() },
                            ) {
                                eprintln!("发送消息事件失败: {}", e);
                            } else {
                                eprintln!("成功发送消息事件到前端，是否为群组消息: {}", is_group_message);
                            }
                        }
                        Err(e) => {
                            eprintln!("解析消息失败: {}, 原始文本: {}", e, text);
                            // 即使解析失败，也发送原始文本
                            let _ = app_clone.emit(
                                "websocket-message",
                                MessageEvent {
                                    message: serde_json::json!({ "raw": text }),
                                },
                            );
                        }
                    }
                }
                Ok(WsMessage::Binary(data)) => {
                    _last_message_time = std::time::Instant::now();
                    consecutive_errors = 0;
                    // 二进制消息，尝试转换为字符串
                    if let Ok(text) = String::from_utf8(data) {
                        eprintln!("收到 WebSocket 二进制消息（已转换为文本）: {}", text);
                        let _ = app_clone.emit(
                            "websocket-message",
                            MessageEvent {
                                message: serde_json::json!({ "raw": text }),
                            },
                        );
                    } else {
                        eprintln!("收到 WebSocket 二进制消息（无法转换为文本）");
                    }
                }
                Ok(WsMessage::Close(_)) => {
                    // 连接关闭
                    eprintln!("WebSocket 连接已关闭");
                    let mut state_guard = state_arc.lock().unwrap();
                    state_guard.sender = None;
                    state_guard.subscription_id = None;

                    let _ = app_clone.emit(
                        "websocket-status",
                        ConnectionEvent {
                            status: "disconnected".to_string(),
                            message: Some("连接已关闭".to_string()),
                        },
                    );
                    break;
                }
                Ok(WsMessage::Ping(_data)) => {
                    _last_message_time = std::time::Instant::now();
                    // 收到 ping，自动回复 pong（tokio-tungstenite 会自动处理）
                    eprintln!("收到 WebSocket Ping，连接正常");
                }
                Ok(WsMessage::Pong(_)) => {
                    _last_message_time = std::time::Instant::now();
                    // 收到 pong，连接正常
                    eprintln!("收到 WebSocket Pong，连接正常");
                }
                Ok(WsMessage::Frame(_)) => {
                    // 忽略 frame 消息（通常由库内部处理）
                }
                Err(e) => {
                    consecutive_errors += 1;
                    eprintln!("WebSocket接收错误 (连续错误 {} 次): {}", consecutive_errors, e);
                    
                    // 如果连续错误太多，认为连接已断开
                    if consecutive_errors >= MAX_CONSECUTIVE_ERRORS {
                        eprintln!("连续错误次数过多，认为连接已断开");
                        let mut state_guard = state_arc.lock().unwrap();
                        state_guard.sender = None;
                        state_guard.subscription_id = None;

                        let _ = app_clone.emit(
                            "websocket-status",
                            ConnectionEvent {
                                status: "error".to_string(),
                                message: Some(format!("连接错误过多，已断开: {}", e)),
                            },
                        );
                        break;
                    }
                    
                    // 检查错误类型，判断是否是连接重置
                    let error_msg = e.to_string();
                    let is_connection_reset = error_msg.contains("Connection reset") 
                        || error_msg.contains("connection reset")
                        || error_msg.contains("Broken pipe")
                        || error_msg.contains("broken pipe")
                        || error_msg.contains("Connection aborted")
                        || error_msg.contains("connection aborted")
                        || error_msg.contains("End of file")
                        || error_msg.contains("end of file");

                    if is_connection_reset {
                        eprintln!("检测到连接重置，断开连接");
                        let mut state_guard = state_arc.lock().unwrap();
                        state_guard.sender = None;
                        state_guard.subscription_id = None;

                        let _ = app_clone.emit(
                            "websocket-status",
                            ConnectionEvent {
                                status: "error".to_string(),
                                message: Some("连接被重置，将尝试重连".to_string()),
                            },
                        );
                        break;
                    }
                    
                    // 对于其他错误，继续尝试接收（可能是临时错误）
                    // 但记录日志以便调试
                }
            }
        }
        
        // 连接断开后的清理
        eprintln!("WebSocket 接收任务已退出，subscription_id: {:?}", subscription_id_clone);
        let mut state_guard = state_arc.lock().unwrap();
        if state_guard.subscription_id.as_ref() == Some(&subscription_id_clone) {
            state_guard.sender = None;
            state_guard.subscription_id = None;
        }
    });

    Ok(())
}

/// 断开 WebSocket 的内部实现
async fn disconnect_websocket_internal(
    app: &AppHandle,
    state: &State<'_, Arc<Mutex<WebSocketState>>>,
) -> Result<(), String> {
    let mut state_guard = state.lock().unwrap();
    if let Some(sender) = state_guard.sender.take() {
        // 发送关闭消息
        let _ = sender.send(WsMessage::Close(None));
        state_guard.subscription_id = None;

        // 发送断开事件
        app.emit(
            "websocket-status",
            ConnectionEvent {
                status: "disconnected".to_string(),
                message: Some("已断开连接".to_string()),
            },
        )
        .map_err(|e| format!("发送事件失败: {}", e))?;
    }
    Ok(())
}

/// 断开 WebSocket
#[tauri::command]
pub async fn disconnect_websocket(
    app: AppHandle,
    state: State<'_, Arc<Mutex<WebSocketState>>>,
) -> Result<(), String> {
    disconnect_websocket_internal(&app, &state).await
}

/// 发送消息（虽然后端可能忽略，但保留接口）
#[tauri::command]
pub async fn send_websocket_message(
    message: String,
    state: State<'_, Arc<Mutex<WebSocketState>>>,
) -> Result<(), String> {
    let state_guard = state.lock().unwrap();
    if let Some(sender) = &state_guard.sender {
        sender
            .send(WsMessage::Text(message))
            .map_err(|e| format!("发送消息失败: {}", e))?;
        Ok(())
    } else {
        Err("WebSocket未连接".to_string())
    }
}

/// 获取连接状态
#[tauri::command]
pub fn get_websocket_status(
    state: State<'_, Arc<Mutex<WebSocketState>>>,
) -> Result<String, String> {
    let state_guard = state.lock().unwrap();
    // 返回实际的连接状态，而不是 subscription_id
    if state_guard.sender.is_some() {
        Ok("connected".to_string())
    } else {
        Ok("disconnected".to_string())
    }
}

